package cn.datawin.routeserver.consumer;

import cn.datawin.routeserver.input.service.InputDealMachineService;
import cn.hutool.extra.spring.SpringUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;

import java.nio.charset.StandardCharsets;

import static cn.datawin.constant.RocketMqConstant.CMPP_INPUT_TOPIC;

@Slf4j
@Configuration
public class CmppInputTopicConsumerConfig {

	private final Environment environment;

	public CmppInputTopicConsumerConfig(Environment environment) {
		this.environment = environment;
	}

	/**
	 * 配置消费者
	 */
	@Bean(initMethod = "start", destroyMethod = "shutdown")
	public DefaultMQPushConsumer cmppInputTopicConsumer() throws Exception {
		String nameServer = environment.getProperty("spring.rocketmq.name-server");
		String consumerGroup = environment.getProperty("spring.rocketmq.consumer.group");

		DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
		consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
		consumer.setNamesrvAddr(nameServer);

		// 订阅主题和标签
		consumer.subscribe(CMPP_INPUT_TOPIC, "*");

		// 设置消息监听
		consumer.registerMessageListener((MessageListenerConcurrently)
				(msgs, context) -> {
//
					for (MessageExt msg : msgs) {
						try {
							String message = new String(msg.getBody(), StandardCharsets.UTF_8);
							log.info("接收到消息: {} ", message );
							// 处理消息业务逻辑
							SpringUtil.getBean(InputDealMachineService.class).dealInputLog(message);
						} catch (Exception e) {
							log.error("处理消息失败", e);
							// 消费失败，返回重试
							return ConsumeConcurrentlyStatus.RECONSUME_LATER;
						}
					}
					return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
				});

		log.info("RocketMQ Consumer 初始化完成，NameServer: {}", nameServer);
		return consumer;
	}
}
